package com.jason.partitioner;

import com.jason.service.IDataInputService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/************************************************************************************************
 *  @Author:zhouhx
 *  @Date: 10:02 2020/11/18                                                                                                                                                 
 *  @Description: //分区处理类
 *  @Version:1.0
 ************************************************************************************************
 **/
@Service
public class DataPartitioner implements Partitioner {

    private static final Logger log = LoggerFactory.getLogger(DataPartitioner.class);

    @Autowired
    private IDataInputService dataInputService;

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {

        log.info("partition  gridsize is " + gridSize);
        Map<String, ExecutionContext> result = new HashMap<>();

        int maxId = dataInputService.getInputMaxId();
        int minId = dataInputService.getInputMinId();

        int targetSize =(maxId-minId)/gridSize+1;
        int number = 0;
        int startIndex = minId;
        int endIndex = startIndex + targetSize -1;

        while (startIndex <= maxId) {
            ExecutionContext value = new ExecutionContext();

            log.info("\nStarting : Thread" + number);
            log.info("fromId : " + startIndex);
            log.info("toId : " + endIndex);

            result.put("partition" + number, value);

            if (endIndex >= maxId) {
                endIndex = maxId;
            }
            value.putInt("fromId", startIndex);
            value.putInt("toId", endIndex);
            startIndex += targetSize;
            endIndex += targetSize;
            number++;
        }

        return result;
    }
}
